package com.dddframework.mq.impl;

import com.dddframework.core.contract.MQEvent;
import com.dddframework.core.contract.exception.ServiceException;
import com.dddframework.core.utils.JsonKit;
import com.dddframework.mq.core.MQClient;
import com.dddframework.mq.core.MQFilter;
import com.dddframework.mq.core.MQListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.StreamEntry;
import redis.clients.jedis.StreamEntryID;
import redis.clients.jedis.StreamGroupInfo;
import redis.clients.jedis.params.XReadGroupParams;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * Redis Stream 客户端实现
 */
@Slf4j(topic = "### BASE-MQ : redisStreamClient ###")
@Component
public final class RedisStreamClient implements MQClient {
    private static Jedis PRODUCER;
    BlockingQueue<MQEvent> SENDING_MSGS = new LinkedBlockingQueue<>();
    private ExecutorService es = Executors.newSingleThreadExecutor();
    private AtomicBoolean started = new AtomicBoolean(false);

    @Override
    public String impl() {
        return "redisStream";
    }

    private void run() {
        es.submit(() -> {
            log.info("Started MQ publisher");
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    MQEvent mqEvent = SENDING_MSGS.take();
                    String message = serialization().serialize(mqEvent);
                    // streamKey=namespace:topic:tag或namespace:topic或topic
                    String streamKey = String.format("%s%s%s", config().namespace(":"), mqEvent.getTopic(),
                            (mqEvent.getTag() == null || mqEvent.getTag().isEmpty()) ? "" : (":" + mqEvent.getTag()));
                    if (PRODUCER == null) {
                        PRODUCER = jedis();
                    }
                    try {
                        PRODUCER.xadd(streamKey, StreamEntryID.NEW_ENTRY, Collections.singletonMap("message", message));
                        log.info("Publish MQ [{}]: {}", streamKey, message);
                    } catch (Exception e) {
                        log.error("Publish MQ [{}]: {} failed!", streamKey, message, e);
                    }
                } catch (Exception e) {
                    log.error("Publish MQ failed!", e);
                }
            }

            log.info("MQ publisher stopped");
        });
    }

    @Override
    public Consumer<MQEvent> initProducer() {
        if (started.compareAndSet(false, true)) {
            run();
        }
        return mqEvent -> SENDING_MSGS.offer(mqEvent);
    }

    @Override
    public boolean initConsumer(MQListener mqListener) throws Exception {
        // streamKey=namespace:topic:tag或namespace:topic或topic
        List<String> topics = new ArrayList<>();
        if (mqListener.getTags() != null && !mqListener.getTags().isEmpty()) {
            Set<String> tags = MQFilter.findIncludes(mqListener.getTags());
            if (!tags.isEmpty()) {
                for (String tag : tags) {
                    topics.add(config().namespace(":") + mqListener.getTopic() + ":" + tag);
                }
            }
        } else {
            topics.add(config().namespace(":") + mqListener.getTopic());
        }
        Jedis jedis = jedis();
        Map<String, StreamEntryID> streamKeys = new HashMap<>();
        // 创建消费组（忽略已存在）
        for (String topic : topics) {
            try {
                List<StreamGroupInfo> streamGroupInfos = jedis.xinfoGroup(topic);
                List<String> streamGroupInfoGroupNames = streamGroupInfos.stream().map(StreamGroupInfo::getName).collect(Collectors.toList());
                if (!streamGroupInfoGroupNames.contains(mqListener.getGroup())) {
                    // 创建消费者组，使用$表示最后一个条目的ID，MKSTREAM选项设置为true以确保如果Stream不存在则创建它
                    jedis.xgroupCreate(topic, mqListener.getGroup(), new StreamEntryID("0-0"), true);
                }
                streamKeys.put(topic, StreamEntryID.UNRECEIVED_ENTRY);
            } catch (Exception e) {
                log.info("Create consumer group failed: key={}, groupname={}", topic, mqListener.getGroup(), e);
            }
        }
        if (streamKeys.isEmpty()) {
            return false;
        }
        new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    // 从消费组中读取消息
                    List<Map.Entry<String, List<StreamEntry>>> messages = jedis.xreadGroup(mqListener.getGroup(),
                            mqListener.getMethod().getName(),
                            XReadGroupParams.xReadGroupParams().count(1).block(1000),
                            streamKeys);
                    if (messages == null || messages.isEmpty()) {
                        // 没有消息，可以选择在这里短暂休眠，避免CPU占用过高
                        TimeUnit.MILLISECONDS.sleep(1000);
                        continue;
                    }

                    for (Map.Entry<String, List<StreamEntry>> entry : messages) {
                        for (StreamEntry streamEntry : entry.getValue()) {
                            String message = streamEntry.getFields().get("message");
                            MQEvent mqEvent = mqListener.getDeserialize().apply(message);
                            try {
                                if (config().isAutoAck()) {
                                    jedis.xack(entry.getKey(), mqListener.getGroup(), streamEntry.getID());
                                }
                                // 消费消息
                                consume(mqListener, mqEvent);
                                // 确认消息
                                if (!config().isAutoAck()) {
                                    jedis.xack(entry.getKey(), mqListener.getGroup(), streamEntry.getID());
                                }
                            } catch (Throwable e) {
                                if (e instanceof ServiceException) {
                                    log.error("Consume MQ [{}] failed: {} => {}", mqListener.namespaceTopicTags(), e.getMessage(), JsonKit.toJson(mqEvent));
                                } else {
                                    log.error("Consume MQ [{}] failed: {}", mqListener.namespaceTopicTags(), mqEvent, e);
                                }
                            }
                        }
                    }
                } catch (Exception e) {
                    log.error("Consume MQ [{}] failed!", mqListener.namespaceTopicTags(), e);
                    try {
                        TimeUnit.MILLISECONDS.sleep(5000);
                    } catch (InterruptedException ex) {
                        throw new RuntimeException(ex);
                    }
                }
            }
        }).start();

        return true;
    }

    private synchronized Jedis jedis() {
        String[] hostAndPort = config().getServer().split(":");
        if (!config().getPassword().isEmpty()) {
            Jedis jedis = new Jedis(hostAndPort[0], Integer.parseInt(hostAndPort[1]), 5000);
            jedis.auth(config().getPassword());
            return jedis;
        } else {
            return new Jedis(hostAndPort[0], Integer.parseInt(hostAndPort[1]), 5000);
        }
    }
}